package com.zyx.hbasedemo;

import org.apache.phoenix.jdbc.PhoenixDriver;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Properties;

/**
 * @author zyx
 * @since 2021/7/4 18:38
 * desc: 利用Phoenix操作HBase实例
 */
public class PhoenixDemo {
    private Connection connection;
    private static final String PHOENIX_URL = "jdbc:phoenix:linux201,linux202,linux203:2181";

    @Before
    public void init() throws Exception {
        // 1、加载驱动
        Class.forName(PhoenixDriver.class.getCanonicalName());
        // 2、获取connnection对象
        // 设置开启namespace mapping
        Properties props = new Properties();
        props.setProperty("phoenix.schema.isNamespaceMappingEnabled", String.valueOf(true));
        props.setProperty("phoenix.schema.mapSystemTablesToNamespace", String.valueOf(true));
        // 设置Phoenix连接的超时时间
        props.setProperty("phoenix.query.timeoutMs", "1200000");
        props.setProperty("hbase.rpc.timeout", "1200000");
        props.setProperty("hbase.client.scanner.timeout.period", "1200000");
        connection = DriverManager.getConnection(PHOENIX_URL, props);
        connection.setAutoCommit(true);
    }

    /**
     * 建表
     */
    @Test
    public void createTable() throws Exception {

        // 3、获取statement对象
        // select * from user where name='zhangsan' or 1=1
        String sql = "create table \"bigdata\".student(" +
                "id varchar primary key," +
                "name varchar," +
                "age varchar)COLUMN_ENCODED_BYTES=0";
        PreparedStatement statement = connection.prepareStatement(sql);
        // 4、执行建表操作
        statement.execute();
        // 5、关闭资源
        statement.close();
    }

    @After
    public void close() throws Exception {
        if (connection != null)
            connection.close();
    }

    /**
     * 插入单条数据
     */
    @Test
    public void insert() throws Exception {
        String sql = "upsert into \"bigdata\".student values(?,?,?)";
        PreparedStatement statement = connection.prepareStatement(sql);

        statement.setString(1, "1001");
        statement.setString(2, "zhangsan");
        statement.setString(3, "30");
        statement.executeUpdate();
        // 手动提交
        // connection.commit();
        statement.close();
    }

    /**
     * 批量插入
     */
    @Test
    public void batchInsert() throws Exception {
        String sql = "upsert into \"bigdata\".student values(?,?,?)";
        PreparedStatement statement = connection.prepareStatement(sql);

        for (int i = 0; i <= 130; i++) {

            statement.setString(1, "10011" + i);
            statement.setString(2, "zhangsan" + i);
            statement.setString(3, "2" + i);
            //将当前数据添加到一个批次中
            statement.addBatch();
            //50条数据一个批次
            if (i % 50 == 0) {
                //提交一个批次的数据
                statement.executeBatch();
                //清空已经提交的批次
                statement.clearBatch();
            }
        }
        //提交最后一个不满50条的批次数据
        statement.executeBatch();

        statement.close();
    }

    /**
     * 删除数据
     */
    @Test
    public void delete() throws Exception {

        String sql = "delete from \"bigdata\".student where id=?";
        PreparedStatement statement = connection.prepareStatement(sql);

        statement.setString(1, "1001");

        statement.executeUpdate();

        statement.close();
    }

    /**
     * 查询数据
     */
    @Test
    public void query() throws Exception {
        String sql = "select * from \"bigdata\".student where id > ?";
        PreparedStatement statement = connection.prepareStatement(sql);

        statement.setString(1, "100114");

        ResultSet resultSet = statement.executeQuery();
        while (resultSet.next()) {
            //获取每一行的对应的字段的值
            String id = resultSet.getString("id");
            String name = resultSet.getString("name");
            String age = resultSet.getString("age");
            System.out.println("name=" + name + ",id=" + id + ",age=" + age);
        }

        statement.close();
    }

    /**
     * 删除表
     *
     * @throws Exception
     */
    @Test
    public void dropTable() throws Exception {
        String sql = "drop table \"bigdata\".student";
        PreparedStatement statement = connection.prepareStatement(sql);
        statement.execute();
        statement.close();
    }
}
